package esmp.bootstrap.bean;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import redis.task.DeferredResultTask;
import redis.task.DeferredResultTaskManager;
import redis.task.DeferredResultTaskResult;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Arrays;

public class RedisEventSubscriber implements MessageListener, InitializingBean {
    private DeferredResultTaskManager taskManager;
    private ObjectMapper objectMapper = new ObjectMapper();
    private byte[] taskChannel;
    private byte[] resultChannel;

    public RedisEventSubscriber(DeferredResultTaskManager taskManager) {
        this.taskManager = taskManager;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        taskChannel = taskManager.getTaskChannel().getBytes();
        resultChannel = taskManager.getResultChannel().getBytes();
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        Charset charset = Charset.forName("UTF-8");
        try {
            if (Arrays.equals(message.getChannel(), taskChannel)) {
                taskManager.executeTask((DeferredResultTask<?>) objectMapper.readValue(new String(message.getBody(), charset), DeferredResultTask.class));
            } else if (Arrays.equals(message.getChannel(), resultChannel)) {
                taskManager.receivedResult((DeferredResultTaskResult<?>) objectMapper.readValue(new String(message.getBody(), charset), DeferredResultTaskResult.class));
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
